package streamproc.buffercomponents;

import java.util.Calendar;
import java.util.HashMap;
import java.util.Vector;

import streamdef.StreamDefinitionManager;
import streamproc.ChainComponent;
import event.Event;
import event.EventFormatManager;

public class BufferManagerComponent extends ChainComponent {
	private static BufferManagerComponent sInstance = null;

	public static BufferManagerComponent instance() {
		if (sInstance == null)
			sInstance = new BufferManagerComponent();

		return sInstance;
	}

	// how much items should ideally be removed after a cleanup
	private final static double desiredPercentageRemovedItemsAfterCleanup = 0.1;
	private HashMap<Long, Buffer> streamIDToEventBuffer = null;
	private Vector<Long> streamIDs = null;

	// reset after cleanup
	private HashMap<Long, Long> streamIDToTimestampNextCleanup = null; // ms
	// incremented after each push, reset after cleanup
	private HashMap<Long, Long> streamIDToNumberOfPushedEvents = null;

	private Event eventToBuffer = null;

	private boolean active = false;

	private BufferManagerComponent() {
		streamIDToEventBuffer = new HashMap<Long, Buffer>();
		streamIDToTimestampNextCleanup = new HashMap<Long, Long>();
		streamIDToNumberOfPushedEvents = new HashMap<Long, Long>();
		streamIDs = new Vector<Long>();
		eventToBuffer = null;
		active = false;
	}

	public void reset() {
		if (streamIDToEventBuffer != null)
			streamIDToEventBuffer.clear();
		if (streamIDToTimestampNextCleanup != null)
			streamIDToTimestampNextCleanup.clear();
		if (streamIDToNumberOfPushedEvents != null)
			streamIDToNumberOfPushedEvents.clear();
		if (streamIDs != null)
			streamIDs.clear();
		if (eventToBuffer != null)
			eventToBuffer = null;

		active = false;
	}

	protected void finalize() {
		reset();
	}

	public void init() {
		reset();

		StreamDefinitionManager pm = StreamDefinitionManager.instance();

		Calendar calendar = Calendar.getInstance();
		java.util.Date now = calendar.getTime();
		java.sql.Timestamp currTimestamp = new java.sql.Timestamp(now.getTime());
		long currTS = currTimestamp.getTime();

		Vector<Long> _streamIDs = EventFormatManager.instance()
				.getValidStreamIDs();
		for (int i = 0; i < _streamIDs.size(); ++i) {
			if (pm.streamIDIsUsedInOtherQueries(_streamIDs.get(i))
					&& pm.streamIDIsUsedInOtherQueriesButMayNotAlwaysOccur(_streamIDs
							.get(i))) {
				Buffer b = new Buffer(_streamIDs.get(i));
				streamIDToEventBuffer.put(new Long(_streamIDs.get(i)), b);
				streamIDs.add(new Long(_streamIDs.get(i)));

				long nextCleanupTS = (long) (currTS + (b.getTotalRange() * 1000.0 * (1.0 + desiredPercentageRemovedItemsAfterCleanup)));

				streamIDToTimestampNextCleanup.put(new Long(_streamIDs.get(i)),
						nextCleanupTS);
				streamIDToNumberOfPushedEvents.put(new Long(_streamIDs.get(i)),
						new Long(0));
			}
		}

		active = true;
	}

	public boolean isActive() {
		return active;
	}

	public synchronized void processEvent(Event e) {
		long sID = e.getStreamID();

		if (streamIDToEventBuffer.get(sID) != null) {
			eventToBuffer = e;
			bufferEvents();
		}

		next.processEvent(e);
	}

	public synchronized void bufferEvents() {
		if (eventToBuffer != null) {
			Event e = eventToBuffer;
			long sID = e.getStreamID();
			if (streamIDToEventBuffer.get(sID) != null) {
				streamIDToEventBuffer.get(sID).processEvent(e); // store it

				// increment #pushes for stream
				long nrPushedEvents = streamIDToNumberOfPushedEvents.get(sID);
				nrPushedEvents++;
				streamIDToNumberOfPushedEvents.put(sID, nrPushedEvents);

				if (nrPushedEvents % 1000 == 0) {
					Buffer b = getBuffer(sID);

					Calendar calendar = Calendar.getInstance();
					java.util.Date now = calendar.getTime();
					java.sql.Timestamp currTimestamp = new java.sql.Timestamp(
							now.getTime());
					long currTS = currTimestamp.getTime();

					if (currTS >= streamIDToTimestampNextCleanup.get(sID)) {
						b.removeOutdatedEventsBinarySearch(currTimestamp);
						long nextTS = (long) (currTS + (b.getTotalRange() * 1000.0 * (desiredPercentageRemovedItemsAfterCleanup)));

						streamIDToTimestampNextCleanup.put(sID, nextTS);
						streamIDToNumberOfPushedEvents.put(sID, new Long(0));
					}
				}
			}
		}
		eventToBuffer = null;
	}

	public Buffer getBuffer(long streamID) {
		return streamIDToEventBuffer.get(streamID);
	}

	public synchronized void cleanBuffers() {
		Calendar calendar = Calendar.getInstance();
		java.util.Date now = calendar.getTime();
		java.sql.Timestamp currTimestamp = new java.sql.Timestamp(now.getTime());
		long currTS = currTimestamp.getTime();

		for (int i = 0; i < this.streamIDs.size(); ++i) {
			long ID = streamIDs.get(i);
			Buffer b = getBuffer(ID);
			b.removeOutdatedEventsBinarySearch(currTimestamp);

			long nextTS = (long) (currTS + (b.getTotalRange() * 1000.0 * (desiredPercentageRemovedItemsAfterCleanup)));

			streamIDToTimestampNextCleanup.put(ID, nextTS);
			streamIDToNumberOfPushedEvents.put(ID, new Long(0));
		}
	}
}
